//
// Created by syj on 2023/10/10.
//

#ifndef HELLO_DDS_DDS_TRANSMITTER_H
#define HELLO_DDS_DDS_TRANSMITTER_H

#include <iostream>
#include <map>
#include <atomic>
#include <condition_variable>
#include <mutex>

#include <fastdds/dds/publisher/DataWriterListener.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>
#include <fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h>
#include <fastdds/rtps/transport/UDPv4TransportDescriptor.h>
#include <fastdds/rtps/transport/UDPv6TransportDescriptor.h>

#include <fastrtps/attributes/ParticipantAttributes.h>
#include <fastrtps/attributes/PublisherAttributes.h>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
#include "transport/transmitter/transmitter.h"
#include "transport/dds/DDSMessagePubSubTypes.h"
#include "transport/dds/DDSMessagePubSub.hpp"

namespace motovis {
    namespace transport {
        template <typename MessageT>
        class DDSTransmitter: public Transmitter<MessageT>
        {
        public:
            using MessagePtr = std::shared_ptr<MessageT>;
            DDSTransmitter()
            : type_(new MessagePubSubType<MessageT>())
            {
                DDSPublisher::Instance()->init("DataWriter");
                init();
            }
            virtual ~DDSTransmitter()
            {
                if (DDSPublisher::Instance()->GetParticipant() != nullptr)
                {
                    if (DDSPublisher::Instance()->GetPublisher() != nullptr)
                    {
//            if (writer_ != nullptr)
                        {
//                DDSPublisher::Instance()->GetPublisher()->delete_datawriter(writer_);
                        }

                        for (auto& item: m_writer_)
                        {
                            DDSPublisher::Instance()->GetPublisher()->delete_datawriter(item.second);
                        }
                        DDSPublisher::Instance()->GetParticipant()->delete_publisher(DDSPublisher::Instance()->GetPublisher());
                    }
//        if (topic_ != nullptr)
                    {
//            DDSPublisher::Instance()->GetParticipant()->delete_topic(topic_);
                    }

                    for (auto& item: m_topic_)
                    {
                        DDSPublisher::Instance()->GetParticipant()->delete_topic(item.second);
                    }
                    eprosima::fastdds::dds::DomainParticipantFactory::get_instance()->delete_participant(DDSPublisher::Instance()->GetParticipant());
                }
            }

            /* channel_name: topic_name
             * msg: 指向数据的指针
             * */
            bool transmit(const std::string &channel_name,
                          const MessagePtr &msg) override
            {
                return creatWriter(channel_name,msg);
            }

            //! Initialize the publisher
            /* domain: domainID
             * num_wait_matched: 每次 writer 需要凑齐的 reader 个数, 否则数据不发布
             * */
            bool init(uint32_t num_wait_matched=0)
            {
                listener_.set_num_wait_matched(num_wait_matched);

                // REGISTER THE TYPE
                type_.register_type(DDSPublisher::Instance()->GetParticipant());
                return true;
            }

            bool creatWriter(const std::string& topic_name, const MessagePtr &msg,
                             bool async=false,
                             DDSTransportType transport=DDSTransportType::SHM,
                             bool reliable=false,
                             bool transient=false,
                             const std::string& partitions = "",
                             bool use_ownership = false,
                             unsigned int ownership_strength=0,
                             const std::string& profile = "")
            {
                if (nullptr == DDSPublisher::Instance()->GetParticipant() || nullptr == DDSPublisher::Instance()->GetPublisher())
                {
                    return false;
                }
                // CREATE THE WRITER param
                eprosima::fastdds::dds::DataWriterQos wqos = eprosima::fastdds::dds::DATAWRITER_QOS_DEFAULT;
                // Data sharing set in endpoint. If it is not default, set it to off
                if (transport != DDSTransportType::DEFAULT)
                {
                    wqos.data_sharing().off();
                }
                else
                {
                    wqos.data_sharing().automatic();  // default
                }
                if (async)
                {
                    wqos.publish_mode().kind = eprosima::fastdds::dds::ASYNCHRONOUS_PUBLISH_MODE;
                }
                else
                {
                    wqos.publish_mode().kind = eprosima::fastdds::dds::SYNCHRONOUS_PUBLISH_MODE;    // default
                }
                if (reliable)
                {
                    wqos.reliability().kind = eprosima::fastdds::dds::RELIABLE_RELIABILITY_QOS;
                    wqos.history().kind = eprosima::fastdds::dds::KEEP_ALL_HISTORY_QOS;
                }
                else
                {
                    wqos.reliability().kind = eprosima::fastdds::dds::BEST_EFFORT_RELIABILITY_QOS;  // default in this example (although default value for
                    // writters' qos actually is RELIABLE)
                }

                if (transient)
                {
                    wqos.durability().kind = eprosima::fastdds::dds::TRANSIENT_LOCAL_DURABILITY_QOS;
                    wqos.history().kind = eprosima::fastdds::dds::KEEP_ALL_HISTORY_QOS;     // store previously sent samples so they can be resent to newly
                    // matched DataReaders
                }
                else
                {
                    wqos.durability().kind = eprosima::fastdds::dds::VOLATILE_DURABILITY_QOS;   // default in this example (although default value for
                    // writters' qos actually is TRANSIENT_LOCAL)
                }
                // Set ownership
                if (use_ownership)
                {
                    wqos.ownership().kind = eprosima::fastdds::dds::OwnershipQosPolicyKind::EXCLUSIVE_OWNERSHIP_QOS;
                    wqos.ownership_strength().value = ownership_strength;
                }

                /* 创建具体的发布端口 */
                eprosima::fastdds::dds::Topic* pTopic{nullptr};
                eprosima::fastdds::dds::DataWriter* pWriter{nullptr};
                auto topicItr = m_topic_.find(topic_name);
                if (topicItr != m_topic_.end())
                {
                    pTopic = topicItr->second;
                    if (pTopic == nullptr)
                    {
                        return false;
                    }
                    auto writerItr = m_writer_.find(pTopic);
                    if (writerItr != m_writer_.end())
                    {
                        pWriter = writerItr->second;
                        if (pWriter == nullptr)
                        {
                            return false;
                        }
                    }
                    else
                    {
                        /* 创建DataWriter */
                        pWriter = DDSPublisher::Instance()->GetPublisher()->create_datawriter(pTopic, wqos, &listener_);
                        if (pWriter == nullptr)
                        {
                            return false;
                        }
                        m_writer_.insert(std::pair<eprosima::fastdds::dds::Topic*,eprosima::fastdds::dds::DataWriter*>{pTopic,pWriter});
                    }
                }
                else
                {
                    // CREATE THE TOPIC
                    pTopic = DDSPublisher::Instance()->GetParticipant()->create_topic(topic_name, "Motovis", eprosima::fastdds::dds::TOPIC_QOS_DEFAULT);
                    if (pTopic == nullptr)
                    {
                        return false;
                    }
                    pWriter = DDSPublisher::Instance()->GetPublisher()->create_datawriter(pTopic, wqos, &listener_);
                    if (pWriter == nullptr)
                    {
                        return false;
                    }
                    m_topic_.insert(std::pair<std::string,eprosima::fastdds::dds::Topic*>{topic_name,pTopic});
                    m_writer_.insert(std::pair<eprosima::fastdds::dds::Topic*,eprosima::fastdds::dds::DataWriter*>{pTopic,pWriter});
                }



                while (!is_stopped())
                {
                    if (listener_.enough_matched())
                    {
                        if (nullptr != pWriter)
                        {
                            pWriter->write(msg.get());
                        }
                        std::this_thread::sleep_for(std::chrono::milliseconds(50));
                    }
                    else
                    {
                        listener_.wait();
                    }
                }

                std::cout << "Publisher Participant created with DataWriter Guid [ " << m_writer_.size() << " ]." << std::endl;
                return true;
            }

            //! Return the current state of execution
            static bool is_stopped()
            {
                return stop_;
            }

            //! Trigger the end of execution
            static void stop()
            {
                stop_ = true;
                PubListener::awake();
            }

        private:
//            eprosima::fastdds::dds::DomainParticipant* DDSPublisher::Instance()->GetParticipant();
//            eprosima::fastdds::dds::Publisher* DDSPublisher::Instance()->GetPublisher();
            eprosima::fastdds::dds::TypeSupport type_;

            std::map<std::string,eprosima::fastdds::dds::Topic*> m_topic_;
            std::map<eprosima::fastdds::dds::Topic*,
                    eprosima::fastdds::dds::DataWriter*> m_writer_;

            //! Member used for control flow purposes
            static std::atomic<bool> stop_;

            class PubListener : public eprosima::fastdds::dds::DataWriterListener
            {
            public:

                PubListener()
                        : matched_(0)
                        , num_wait_matched_(0)
                {
                }

                ~PubListener() override
                {
                }

                void on_publication_matched(
                        eprosima::fastdds::dds::DataWriter* writer,
                        const eprosima::fastdds::dds::PublicationMatchedStatus& info) override

                {
                    if (info.current_count_change == 1)
                    {
                        matched_ = info.current_count; /* 当前订阅的个数 */
                        std::cout << info.current_count << " Publisher matched [ " << iHandle2GUID(info.last_subscription_handle) << " ]." << std::endl;
                        /* 发送数据成功 */
                        if (enough_matched())
                        {
                            awake();
                        }
                    }
                    else if (info.current_count_change == -1) /* 发布结束了,退出发布接口 */
                    {
                        matched_ = info.current_count;
                        std::cout << info.current_count << " Publisher unmatched [ " << iHandle2GUID(info.last_subscription_handle) << " ]." << std::endl;
                        DDSTransmitter::stop();
                    }
                    else
                    {
                        std::cout << info.current_count_change
                                  << " is not a valid value for PublicationMatchedStatus current count change" << std::endl;
                    }
                }

                //! Set the number of matched DataReaders required for publishing
                void set_num_wait_matched(
                        uint32_t num_wait_matched)
                {
                    num_wait_matched_ = num_wait_matched;
                }

                //! Return true if there are at least num_wait_matched_ matched DataReaders
                bool enough_matched()
                {
                    return matched_ >= num_wait_matched_;
                }

                //! Block the thread until enough DataReaders are matched
                void wait()
                {
                    std::unique_lock<std::mutex> lck(wait_matched_cv_mtx_);
                    wait_matched_cv_.wait(lck, [this]
                    {
                        return enough_matched() || is_stopped();
                    });
                }

                //! Unblock the thread so publication of samples begins/resumes
                static void awake()
                {
                    wait_matched_cv_.notify_all();
                }

            private:
                //! Number of DataReaders matched to the associated DataWriter
                std::atomic<std::uint32_t> matched_; /* topic对应 reader 的个数 */
//        bool transstop;

                //! Number of matched DataReaders required for publishing
                uint32_t num_wait_matched_; /* 需要等待的 reader 的个数 */

                //! Protects wait_matched condition variable
                static std::mutex wait_matched_cv_mtx_;

                //! Waits until enough DataReaders are matched
                static std::condition_variable wait_matched_cv_;
            }
                    listener_;
        };
        template <typename MessageT>
        std::atomic<bool> DDSTransmitter<MessageT>::stop_(false);
        template <typename MessageT>
        std::mutex DDSTransmitter<MessageT>::PubListener::wait_matched_cv_mtx_;
        template <typename MessageT>
        std::condition_variable DDSTransmitter<MessageT>::PubListener::wait_matched_cv_;
    }
}


#endif //HELLO_DDS_DDS_TRANSMITTER_H
